package cn.nawang.ebeim.client.transfer;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpContentDecompressor;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

public class TransferClient {

    private static final Logger LOG = LoggerFactory.getLogger(TransferClient.class);

    private static final int HEARTBEAT = 10;

    private EventLoopGroup eventLoopGroup;

    /**
     * 为了复用EventLoopGroup 、Selector等nio资源，client维持单例,
     * 所有的传输任务,只是新建链路channel,并由netty注册到Selector上
     */
    private static TransferClient transferClient;

    private TransferClient() {
        eventLoopGroup = new NioEventLoopGroup();
    }

    public static TransferClient getTransferClient() {
        if (transferClient == null) {
            synchronized (TransferClient.class) {
                if (transferClient == null) {
                    transferClient = new TransferClient();
                }
            }
        }
        return transferClient;
    }

    public void close() {
        if (eventLoopGroup != null && !eventLoopGroup.isShuttingDown()) {
            try {
                eventLoopGroup.shutdownGracefully().sync();
                LOG.info("EventLoopGroup shutdown complete!");
            } catch (InterruptedException e) {
                LOG.error("EventLoopGroup shutdown error!",e);
            }
        }
    }

    public void download(TransferObject obj, String host, int port,TransferCallback callback){
        final TransferDownloadHandler handler = new TransferDownloadHandler(obj,callback);
        try {
            Bootstrap b = new Bootstrap();
            b.group(eventLoopGroup).channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast("codec", new HttpClientCodec());
                            pipeline.addLast("ping", new IdleStateHandler(HEARTBEAT, 0, 0, TimeUnit.SECONDS));
                            pipeline.addLast("inflater", new HttpContentDecompressor());
                            pipeline.addLast("chunkWriter", new ChunkedWriteHandler());
                            pipeline.addLast("handler", handler);
                        }
                    });
            ChannelFuture f =b.connect(host, port).sync();
            f.channel().closeFuture().sync();
        }catch (Exception e) {
            LOG.error("download error:"+obj,e);
            callback.onTransferFailed(obj,e);
        }
    }

    public void upload(TransferObject obj, String host, int port,TransferCallback callback){
        if (obj.isBreakPointUpload()){
            breakPointUpload(obj,host,port,callback);
        }else {
            normalUpload(obj,host,port,callback,0);
        }
    }

    public void breakPointUpload(TransferObject obj, String host, int port,TransferCallback callback){
        EventLoopGroup elg  = new NioEventLoopGroup();
        final TransferQueryHandler handler = new TransferQueryHandler(obj);
        try {
            Bootstrap b = new Bootstrap();
            b.group(elg).channel(NioSocketChannel.class);
            b.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();
                    pipeline.addLast("codec", new HttpClientCodec());
                    pipeline.addLast("http-encoder", new ChunkedWriteHandler());
                    pipeline.addLast("ping",new IdleStateHandler(0, HEARTBEAT, 0, TimeUnit.SECONDS));
                    pipeline.addLast("handler", handler);
                }
            });
            ChannelFuture f = b.connect(host, port).sync();
            f.channel().closeFuture().sync();
        } catch (Exception e) {
            LOG.error("breakPointUpload error:"+obj,e);
            callback.onTransferFailed(obj,e);
        }finally {
            elg.shutdownGracefully();
            if (handler.isConnect() && handler.isSuccess()) {
                this.normalUpload(obj,host,port,callback,handler.getUploadLength());
            }else {
                callback.onTransferFailed(obj,new TransferException("传输失败，网络未连接"));
            }
        }
    }

    public void normalUpload(TransferObject obj, String host, int port, TransferCallback callback, long preUploadLength){
        final TransferUploadHandler handler = new TransferUploadHandler(obj,callback,preUploadLength);
        try {
            Bootstrap b = new Bootstrap();
            b.group(eventLoopGroup).channel(NioSocketChannel.class);
            b.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();
                    pipeline.addLast("codec", new HttpClientCodec());
                    pipeline.addLast("http-encoder", new ChunkedWriteHandler());
                    pipeline.addLast("ping",new IdleStateHandler(0, HEARTBEAT, 0, TimeUnit.SECONDS));
                    pipeline.addLast("handler", handler);
                }
            });
            ChannelFuture f =b.connect(host, port).sync();
            f.channel().closeFuture().sync();
        } catch (Exception e) {
            LOG.error("normalUpload error:"+obj,e);
            callback.onTransferFailed(obj,e);
        }
    }

}
